package com.example.delay.factory;

import com.example.delay.job.DelayJob;
import com.example.delay.job.RunJob;
import com.example.delay.listener.DefaultJobListener;
import com.example.delay.listener.JobListener;
import com.example.service.HazelcastService;
import com.example.utils.IdUtils;
import com.hazelcast.scheduledexecutor.IScheduledExecutorService;
import com.hazelcast.scheduledexecutor.IScheduledFuture;
import com.hazelcast.topic.ITopic;
import com.hazelcast.topic.Message;
import com.hazelcast.topic.MessageListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/*
 * @className: HazelcastDelayFactory
 * @author: chentao
 * @date: 2025/2/20 上午10:36
 * @Version: 1.0
 * @description: 延迟任务工程
 */
public class HazelcastDelayJobFactory {

    Logger logger = LoggerFactory.getLogger(getClass());

    private static final String DELAY_JOB_TOPIC_KEY = "delay_job_topic_";

    private String name;

    //清理线程 默认10个
    private int core = 10;

    //清理间隔 默认10秒一次
    private int sec = 10;

    private HazelcastService hazelcastService;

    private IScheduledExecutorService scheduledExecutorService;

    private Map<String, IScheduledFuture<Object>> map = new ConcurrentHashMap<>();

    private ITopic<String> topic;

    private JobListener jobListener;

    private ScheduledExecutorService cleanScheduledService;

    public HazelcastDelayJobFactory(String name, HazelcastService hazelcastService) {
        this.hazelcastService = hazelcastService;
        this.name = name;
        this.jobListener = new DefaultJobListener();
        this.topic = hazelcastService.getTopic(DELAY_JOB_TOPIC_KEY + name);
        this.topic.addMessageListener(new MessageListener<String>() {
            @Override
            public void onMessage(Message<String> message) {
                String jobId = message.getMessageObject();
                IScheduledFuture<Object> future = map.remove(jobId);
                if (future != null && !future.isDone()) {
                    future.cancel(false);
                }
            }
        });
        this.scheduledExecutorService = hazelcastService.getScheduledExecutorService(name);
        this.cleanScheduledService = Executors.newScheduledThreadPool(core);
    }

    public HazelcastDelayJobFactory(String name, JobListener jobListener, HazelcastService hazelcastService) {
        this.hazelcastService = hazelcastService;
        this.name = name;
        this.jobListener = jobListener;
        this.topic = hazelcastService.getTopic(DELAY_JOB_TOPIC_KEY + name);
        this.topic.addMessageListener(new MessageListener<String>() {
            @Override
            public void onMessage(Message<String> message) {
                String jobId = message.getMessageObject();
                IScheduledFuture<Object> future = map.remove(jobId);
                if (future != null && !future.isDone()) {
                    future.cancel(false);
                }
            }
        });
        this.scheduledExecutorService = hazelcastService.getScheduledExecutorService(name);
        this.cleanScheduledService = Executors.newScheduledThreadPool(core);
    }

    /****
     *
     * 自动清理线程
     */
    public void start() {
        this.cleanScheduledService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                Iterator<Map.Entry<String, IScheduledFuture<Object>>> iterator = map.entrySet().iterator();
                while (iterator.hasNext()) {
                    Map.Entry<String, IScheduledFuture<Object>> next = iterator.next();
                    if (next.getValue().isDone()) {
                        iterator.remove();
                    }
                }
            }
        }, 0, sec, TimeUnit.SECONDS);
    }

    /***
     * 设置一个定时任务
     * @param jobClass
     * @param delayTime
     * @param timeUnit
     * @return
     * @throws Exception
     */
    public String put(Class<? extends DelayJob> jobClass, long delayTime, TimeUnit timeUnit) throws Exception {
        String jobId = IdUtils.getId();
        RunJob runJob = new RunJob(jobId, jobClass, jobListener);
        IScheduledFuture future = this.scheduledExecutorService.schedule(runJob, delayTime, timeUnit);
        this.map.put(jobId, future);
        return jobId;
    }

    /***
     * 取消延迟任务
     * @return
     */
    public void cancel(String id) {
        if (this.map.containsKey(id)) {
            IScheduledFuture<Object> future = map.remove(id);
            if (future != null && !future.isDone()) {
                future.cancel(false);
            }
        } else {
            this.topic.publish(id);
        }
    }

    /****
     * 关闭并清理资源
     * @throws Exception
     */
    public void stop() {
        this.cleanScheduledService.shutdownNow();
        this.scheduledExecutorService.shutdown();
    }
}
